package com.onpositive.data.serializer.readers;

import java.util.BitSet;

import com.onpositive.data.FilterTester;


public class DoubleAggregator {

	public static final int SUM_MODE = FilterTester.AGR_SUM;
	public static final int AVERAGE_MODE = FilterTester.AGR_AVE;
	public static final int MIN_MODE = FilterTester.AGR_MIN;
	public static final int MAX_MODE = FilterTester.AGR_MAX;
	
	protected int mode = DoubleAggregator.SUM_MODE;

	
	public DoubleAggregator(int op) {
		super();
		this.mode=op;
	}

	public Object getAggregatedValue(double[] values, BitSet ss, BitSet ss1) {
		if (values.length > 0) {
			switch (this.mode) {
			case SUM_MODE:
				return this.computeDoubleSum(values, ss, ss1);
			case AVERAGE_MODE:
				boolean nulls = true;
				{
					int count = 0;
					int a = 0;

					for (double object : values) {
						if (ss.get(a) || ss1.get(a)) {
							a++;
							continue;
						}
						if (object==Double.NaN){
							continue;
						}
						nulls = false;
						count++;
						a++;
					}
					if (nulls) {
						return null;
					}
					double sum = this.computeDoubleSum(values, ss, ss1);
					return sum / count;
				}
			case MIN_MODE: {
				{
					double min = Double.MAX_VALUE;
					int a = 0;
					for (double i : values) {
						if (ss.get(a) || ss1.get(a)) {
							a++;
							continue;
							
						}
						if (i==Double.NaN){
							continue;
						}
						if (i < min) {
							min = i;
						}
						a++;
					}
					if (min == Double.MAX_VALUE) {
						return null;
					}
					return min;
				}
			}
			case MAX_MODE: {
				double min = Double.MIN_VALUE;
				int a = 0;
				for (double i : values) {
					if (ss.get(a) || ss1.get(a)) {
						a++;
						continue;
					}
					if (i==Double.NaN){
						continue;
					}
					if (i> min) {
						min = i;
					}
					a++;
				}
				if (min == Double.MAX_VALUE) {
					return null;
				}
				return min;
			}
			default:
				throw new IllegalStateException(
						"Invalid aggregate calculation mode - " + this.mode);
			}
		}

		return 0;
	}

	protected int computeDoubleSum(double[] values, BitSet s1, BitSet s2) {
		int sum = 0;
		int a = 0;
		for (double value : values) {
			if (s1.get(a) || s2.get(a)) {
				a++;
				continue;
			}
			if (value==Double.NaN){
				continue;
			}
			a++;
			sum += value;
		}
		return sum;
	}

}
